package com.company.nuwa.kafka.config;


import com.company.nuwa.common.utils.IpUtils;
import com.google.common.collect.Maps;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;

import java.util.Map;

/**
 * <p>Kafka配置</p>
 *
 * @author lili
 * @version 1.0: ReceiverConfig.java
 * @date 2019/12/21 16:16
 */
@EnableKafka
@Configuration
public class ReceiverConfig {

    @Value("${spring.kafka.bootstrap-servers:127.0.0.1:9092}")
    private String bootstrapServers;
    @Value("${spring.kafka.consumer.max-poll-records:200}")
    private Integer maxPollRecordsConfig;
    @Value("${spring.kafka.consumer.auto-commit-interval:100}")
    private Integer autoCommitInterval;
    @Value("${spring.kafka.consumer.session-timeout-ms:15000}")
    private String sessionTimeoutMs;
    @Value("${spring.kafka.consumer.auto-offset-reset:earliest}")
    private String autoOffsetReset;
    @Value("${spring.kafka.consumer.group-id:defaultGroup}")
    private String groupId;
    @Value("${server.port:0}")
    private String serverPort;

    private Map<String, Object> consumerConfigs(boolean auto, boolean isQueue) {
        Map<String, Object> props = Maps.newHashMapWithExpectedSize(2 << 2);
        if (auto) {
            props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
            props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);

        } else {
            props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        }
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeoutMs);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecordsConfig);
        if (isQueue) {
            // 队列
            props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        } else {
            // 广播
            String GROUP_ID = "customer-" + IpUtils.getIp() + "-" + serverPort;
            props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        }
        return props;
    }

    /**
     * 1-1手动队列
     *
     * @return
     */
    @Bean("manualListenerContainerFactory")
    public ConcurrentKafkaListenerContainerFactory manualListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory(consumerConfigs(false, true)));
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        factory.setConcurrency(Runtime.getRuntime().availableProcessors());
        factory.setBatchListener(true);
        return factory;
    }

    /**
     * 1-1自动队列
     *
     * @return
     */
    @Bean("autoListenerContainerFactory")
    public ConcurrentKafkaListenerContainerFactory autoListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory(consumerConfigs(true, false)));
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.BATCH);
        factory.setConcurrency(Runtime.getRuntime().availableProcessors());
        factory.setBatchListener(true);
        return factory;
    }

    /**
     * 1-1手动广播
     *
     * @return
     */
    @Bean("manualBroadcastListenerContainerFactory")
    public ConcurrentKafkaListenerContainerFactory manualBroadcastListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory(consumerConfigs(true, true)));
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        factory.setConcurrency(Runtime.getRuntime().availableProcessors());
        factory.setBatchListener(true);
        return factory;
    }

    /**
     * 1-多自动广播
     *
     * @return
     */
    @Bean("autoBroadcastListenerContainerFactory")
    public ConcurrentKafkaListenerContainerFactory autoBroadcastListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory(consumerConfigs(true, false)));
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.BATCH);
        factory.setConcurrency(Runtime.getRuntime().availableProcessors());
        factory.setBatchListener(true);
        return factory;
    }
}